import org.apache.spark.sql.streaming.OutputMode
import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}

class StructuredStreaming {
  def main(args: Array[String]): Unit = {
    //创建数据集
    val spark = SparkSession.builder()
      .master("local[6]")
      .appName("Test")
      .getOrCreate()
    import spark.implicits._
    val source: DataFrame = spark.readStream
      .format("socket")
      .option("host", "192.168.0.155")
      .option("port", 9999)
      .load()
    val sourceDS: Dataset[String] = source.as[String]
    //不输出日志文件
    spark.sparkContext.setLogLevel("WARN")
    //处理
    val words = sourceDS.flatMap(_.split(" "))
      .map((_, 1))
      .groupByKey(_._1)
      .count()
    //展示
    words.writeStream
      //输出的结果为全局的
      .outputMode(OutputMode.Complete())
      //输出到控制台
      .format("console")
      .start()
      //阻塞主线程
      .awaitTermination()
  }
}
